package org.bwf.study.servcie.impl;

import com.baomidou.mybatisplus.core.conditions.Wrapper;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import lombok.extern.slf4j.Slf4j;
import org.bwf.obj.bo.OrderCreateBO;
import org.bwf.obj.bo.SelectQuerySiteBO;
import org.bwf.study.mapper.OrdersMapper;
import org.bwf.study.model.Orders;
import org.bwf.study.model.UserInfo;
import org.bwf.study.model.WatchTimesSiteColums;
import org.bwf.study.servcie.OrderService;
import org.bwf.study.util.JsonUtil;
import org.bwf.study.util.MD5;
import org.bwf.study.util.RedisUntil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;

@Service
@Slf4j
public class OrderServiceImpl implements OrderService {

    @Autowired
    private OrdersMapper ordersMapper;

    @Resource
    private RedisUntil redisUntil;

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @Override
    public Integer SelectSite(SelectQuerySiteBO selectQuerySiteBO) {
        // 如果生成订单后是删除缓存中的座位信息，这里就要进行数据库判断了
        // Order表中的座位信息是Json数组 ，多个座位， 选择的座位只会是一个
        LambdaQueryWrapper<Orders> ordersLambdaQueryWrapper = Wrappers.lambdaQuery();
        ordersLambdaQueryWrapper.eq(Orders::getOrderCinemaId, selectQuerySiteBO.getOrderCinemaId())
                .eq(Orders::getOrderFilmId, selectQuerySiteBO.getOrderFilmId())
                .eq(Orders::getOrderWtId, selectQuerySiteBO.getOrderWtId());
        List<Orders> ordersList = ordersMapper.selectList(ordersLambdaQueryWrapper);
        for(var o : ordersList){
            // 先把每一个订单 o中的座位json数组信息反序列化
            List<WatchTimesSiteColums> orderSiteList =  JsonUtil.decodeJsonToList(o.getOrderSites(), WatchTimesSiteColums.class);
            // 还要把BO中传进来的一个座位json字符串反序列化
            WatchTimesSiteColums selectSiteInfo = JsonUtil.decodeJson(selectQuerySiteBO.getSelectSiteInfo(),WatchTimesSiteColums.class);
            // 再当前订单o对象中反序列化的座位列表信息中过滤， 每一个os的座位no要等于选择座位中的座位no
            List<WatchTimesSiteColums> tmpList = orderSiteList.stream().
                    filter(os -> os.getSite_no().equals(selectSiteInfo.getSite_no())).collect(Collectors.toList());
            if(tmpList.size() > 0){
                log.warn("选择的座位已经被购买.......");
                return null;
            }
        }
        // 处理Redis   "{"site_no",\"1-10-1\"}"
        String key = new MD5().md5(String.format("SelectSite-%s",
                selectQuerySiteBO.getOrderWtId() + selectQuerySiteBO.getSelectSiteInfo().replace(" ","")));
        log.info(key);
        Object obj = redisUntil.get(key);
        if(obj != null){
            log.warn("座位已经被选择........");
            return null;
        }
        //把我们的选座信息放入Redis
        redisUntil.set(key, selectQuerySiteBO, 300);


        return 0;
    }

    @Override
    public Integer OrderCreate(OrderCreateBO orderCreateBO) {

        // 根据orders中的座位列表，查询Redis，需要座位和Redis中的信息对应，还要对应到用户
        List<WatchTimesSiteColums> watchTimesSiteColumsList = JsonUtil.decodeJsonToList(orderCreateBO.getOrderSites(), WatchTimesSiteColums.class);
        for(var site : watchTimesSiteColumsList){
            // 要生成去访问Redis的key
            // site就是单个座位的对象，需要序列化成json字符串，再和场次进行拼接
            String key = new MD5().md5(String.format("SelectSite-%s",
                    orderCreateBO.getOrderWtId() + JsonUtil.encodeObject(site).replace(" ","")));
            Object selectObj = redisUntil.get(key);
            if(selectObj == null){
                log.error("OrderCreate--SiteERROR订单信息中的座位非法.......");
                return null;
            }

            // 验证里面的座位是否是你选择的
            SelectQuerySiteBO selectQuerySiteBO = (SelectQuerySiteBO) selectObj;
            if(!selectQuerySiteBO.getUserToken().equals(orderCreateBO.getUserToken())){
                log.error(("OrderCreate--tokenERROR 用户token非法......."));
                return null;
            }
        }

        Object userObj = redisUntil.get(orderCreateBO.getUserToken());

        if(userObj == null) return null;

        UserInfo userInfo = (UserInfo) userObj;

        Orders orders = new Orders();
        orders.setOrderNo(orderCreateBO.getOrderNo());
        orders.setOrderTime(orderCreateBO.getOrderTime());
        orders.setOrderUserId(userInfo.getUserId());
        orders.setOrderUserNick(userInfo.getUserNickName());
        orders.setOrderCinemaId(orderCreateBO.getOrderCinemaId());
        orders.setOrderCinemaName(orderCreateBO.getOrderCinemaName());
        orders.setOrderFilmId(orderCreateBO.getOrderFilmId());
        orders.setOrderFilmName(orderCreateBO.getOrderFilmName());
        orders.setOrderWdDate(orderCreateBO.getOrderWdDate());
        orders.setOrderWtBegintime(orderCreateBO.getOrderWtBegintime());
        orders.setOrderWtEndtime(orderCreateBO.getOrderWtEndtime());
        orders.setOrderCost(orderCreateBO.getOrderCost());
        orders.setOrderWtHalls(orderCreateBO.getOrderWtHalls());
        orders.setOrderSites(orderCreateBO.getOrderSites());
        orders.setOrderWtId(orderCreateBO.getOrderWtId());
        orders.setOrderState((byte)0);


        kafkaTemplate.send("OrderInfo",JsonUtil.encodeObject(orders));

        return 0;
    }
}
